package com.viknix.threadpool.manager.server.dependence.monitor;

import com.viknix.threadpool.manager.common.util.ClassUtil;
import com.viknix.threadpool.manager.server.dependence.exception.ServerProcessException;
import com.viknix.threadpool.manager.server.dependence.monitor.event.Event;
import com.viknix.threadpool.manager.server.dependence.monitor.event.SlowEvent;
import com.viknix.threadpool.manager.server.dependence.monitor.listeners.Listener;
import com.viknix.threadpool.manager.server.dependence.monitor.listeners.SmartListener;
import com.viknix.threadpool.manager.server.dependence.monitor.publisher.DefaultPublisher;
import com.viknix.threadpool.manager.server.dependence.monitor.publisher.DefaultSharePublisher;
import com.viknix.threadpool.manager.server.dependence.monitor.publisher.EventPublisher;
import com.viknix.threadpool.manager.server.dependence.util.ClassUtils;
import com.viknix.threadpool.manager.server.dependence.util.MapUtils;
import com.viknix.threadpool.manager.server.dependence.util.ThreadUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;

import static com.viknix.threadpool.manager.server.dependence.exception.ServerProcessException.SERVER_ERROR;

/**
 * @Author: Dongqi
 * @Date: 2021/11/1 19:28
 * @Version 1.0
 * @Description: 通知中心
 */
@Slf4j
public class NotifyCenter {

    public static int ringBufferSize = 16384;

    public static int shareBufferSize = 1024;

    private static final AtomicBoolean CLOSED = new AtomicBoolean(false);

    /**
     * 用于生产 DefaultPublisher
     */
    private static BiFunction<Class<? extends Event>, Integer, EventPublisher> publisherFactory = null;

    private static final NotifyCenter INSTANCE = new NotifyCenter();

    /**
     * 用于处理 SlowEvent 事件的 publisher，与常规事件不同，SlowEvent 只有一个 publisher
     */
    private DefaultSharePublisher sharePublisher;

    private static Class<? extends EventPublisher> clazz = null;


    /**
     * 用于存放常规事件多播器：key：事件的全限定类名；value：事件多播器
     */
    private final Map<String, EventPublisher> publisherMap = new ConcurrentHashMap(16);

    private NotifyCenter() {
    }

    static {
        // 内部 ArrayBlockingQueue 缓冲区大小，对于写入吞吐量较高的应用，该值需要适当增加。默认值为 16384
        String ringBufferSizeProperty = "tpm.core.notify.ring-buffer-size";
        ringBufferSize = Integer.getInteger(ringBufferSizeProperty, 16384);

        // 公共发布者的消息暂存队列缓冲区的大小
        String shareBufferSizeProperty = "tpm.core.notify.share-buffer-size";
        shareBufferSize = Integer.getInteger(shareBufferSizeProperty, 1024);

        // 留有扩展点，可以自己根据api机制注入自己的 EventPublisher
        final ServiceLoader<EventPublisher> loader = ServiceLoader.load(EventPublisher.class);
        Iterator<EventPublisher> iterator = loader.iterator();

        if (iterator.hasNext()) {
            clazz = iterator.next().getClass();
        } else {
            clazz = DefaultPublisher.class;
        }

        // 构建 publisher 工厂
        publisherFactory = new BiFunction<Class<? extends Event>, Integer, EventPublisher>() {
            @Override
            public EventPublisher apply(Class<? extends Event> cls, Integer buffer) {
                try {
                    EventPublisher publisher = clazz.newInstance();
                    publisher.init(cls, buffer);
                    return publisher;
                } catch (Throwable ex) {
                    log.error(">>> [😒 Service class newInstance has error] : {}", ex);
                    throw new ServerProcessException(SERVER_ERROR, ex);
                }
            }
        };

        try {
            // 创建并初始化 DefaultSharePublisher 实例。
            INSTANCE.sharePublisher = new DefaultSharePublisher();
            INSTANCE.sharePublisher.init(SlowEvent.class, shareBufferSize);
        } catch (Throwable ex) {
            log.error(">>> [😒 Service class newInstance has error] : {}", ex);
        }

        // 在JVM关闭时会回调这个任务
        ThreadUtils.addShutdownHook(new Runnable() {
            @Override
            public void run() {
                shutdown();
            }
        });
    }

    /**
     * 关闭 Notifycenter 中的多个 publisher
     */
    public static void shutdown() {
        if (!CLOSED.compareAndSet(false, true)) {
            return;
        }
        log.warn(">>> [😐 NotifyCenter] Start destroying Publisher");

        for (Map.Entry<String, EventPublisher> entry : INSTANCE.publisherMap.entrySet()) {
            try {
                EventPublisher eventPublisher = entry.getValue();
                eventPublisher.shutdown();
            } catch (Throwable e) {
                log.error(">>> [😭 EventPublisher] shutdown has error : {}", e);
            }
        }

        try {
            INSTANCE.sharePublisher.shutdown();
        } catch (Throwable e) {
            log.error(">>> [😒 SharePublisher] shutdown has error : {}", e);
        }

        log.warn(">>> [😐 NotifyCenter] Destruction of the end");
    }

    /**
     * 将监听器注册到对应的事件多播器上
     *
     * @param listener 监听器
     */
    public static void registerSubscriber(Listener listener) {

        // 判断该监听者是否监听了多个事件，如果监听了多个事件，那么会循环判断监听的事件类型
        if (listener instanceof SmartListener) {
            for (Class<? extends Event> subscribeType : ((SmartListener) listener).subscribeTypes()) {
                // For case, producer: defaultSharePublisher -> consumer: smartSubscriber.
                if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
                    INSTANCE.sharePublisher.addSubscriber(listener, subscribeType);
                } else {
                    // For case, producer: defaultPublisher -> consumer: subscriber.
                    addSubscriber(listener, subscribeType);
                }
            }
            return;
        }

        // 获取该监听器订阅的事件类型
        final Class<? extends Event> subscribeType = listener.subscribeType();
        // 确定 SlowEvent 是否与指定参数表示的类或接口相同，或者是其超类或超接口。
        if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
            INSTANCE.sharePublisher.addSubscriber(listener, subscribeType);
            return;
        }
        // 将监听器注册到对应的事件多播器上
        addSubscriber(listener, subscribeType);
    }

    /**
     * 将监听器注册到对应的事件多播器上
     *
     * @param listener      监听器
     * @param subscribeType 事件源Class对象
     */
    private static void addSubscriber(Listener listener, Class<? extends Event> subscribeType) {

        final String topic = ClassUtil.getCanonicalName(subscribeType);
        synchronized (NotifyCenter.class) {
            // MapUtils.computeIfAbsent 是非线程安全的
            // 如果该多播器不存在，会通过多播工厂新生产一个多播器
            MapUtils.computeIfAbsent(INSTANCE.publisherMap, topic, publisherFactory, subscribeType, ringBufferSize);
        }
        // 获取事件对应的多播器对象
        EventPublisher publisher = INSTANCE.publisherMap.get(topic);
        // 将监听器注册到该多播器当中
        publisher.addSubscriber(listener);
    }

    /**
     * 注销监听器。
     *
     * @param listener 监听器对象
     */
    public static <T> void deregisterSubscriber(final Listener listener) {
        if (listener instanceof SmartListener) {
            for (Class<? extends Event> subscribeType : ((SmartListener) listener).subscribeTypes()) {
                if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
                    INSTANCE.sharePublisher.removeSubscriber(listener, subscribeType);
                } else {
                    removeSubscriber(listener, subscribeType);
                }
            }
            return;
        }

        final Class<? extends Event> subscribeType = listener.subscribeType();
        if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
            INSTANCE.sharePublisher.removeSubscriber(listener, subscribeType);
            return;
        }

        if (removeSubscriber(listener, subscribeType)) {
            return;
        }
        throw new NoSuchElementException("The subscriber has no event publisher");
    }

    /**
     * 从 publisher 上移除监听器
     *
     * @param listener      监听器对象
     * @param subscribeType 订阅的事件类型.
     * @return 是否成功删除订阅者。
     */
    private static boolean removeSubscriber(final Listener listener, Class<? extends Event> subscribeType) {

        final String topic = ClassUtils.getCanonicalName(subscribeType);
        EventPublisher eventPublisher = INSTANCE.publisherMap.get(topic);
        if (eventPublisher != null) {
            eventPublisher.removeSubscriber(listener);
            return true;
        }
        return false;
    }

    /**
     * 发布事件
     *
     * @param event 需要发布的事件
     * @return 是否发布成功
     */
    public static boolean publishEvent(Event event) {
        try {
            return publishEvent(event.getClass(), event);
        } catch (Throwable ex) {
            log.error(">>> [😒 There was an exception to the message publishing] :: {}", ex);
            return false;
        }
    }

    /**
     * 根据事件的类型发布事件
     *
     * @param eventType 事件的类型
     * @param event     事件对象
     * @return 是否发布成功
     */
    private static boolean publishEvent(Class<? extends Event> eventType, Event event) {

        // 如果是 SlowEvent，使用 sharePublisher 发布事件
        if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
            return INSTANCE.sharePublisher.publish(event);
        }

        // 获取事件的全限定类名
        final String topic = ClassUtil.getCanonicalName(eventType);
        // 从 publisherMap 中获取 publisher
        EventPublisher publisher = INSTANCE.publisherMap.get(topic);
        if (publisher != null) {
            // 通过 publisher 发布事件
            return publisher.publish(event);
        }
        log.warn(">>> 😒 There are no [{}] publishers for this event, please register", topic);
        return false;
    }

    /**
     * Register to share-publisher.
     *
     * @param eventType class Instances type of the event type.
     * @return share publisher instance.
     */
    public static EventPublisher registerToSharePublisher(final Class<? extends SlowEvent> eventType) {
        return INSTANCE.sharePublisher;
    }

    /**
     * 注册 publisher，如果该 publisher 已经被实例化，那么直接返回该 publisher
     *
     * @param eventType    publisher关注的事件类型
     * @param queueMaxSize 发布者的队列最大容量。
     */
    public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) {
        if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
            return INSTANCE.sharePublisher;
        }

        final String topic = ClassUtils.getCanonicalName(eventType);
        synchronized (NotifyCenter.class) {
            // MapUtils.computeIfAbsent is a unsafe method.
            MapUtils.computeIfAbsent(INSTANCE.publisherMap, topic, publisherFactory, eventType, queueMaxSize);
        }
        return INSTANCE.publisherMap.get(topic);
    }

    /**
     * 注销发布者。
     *
     * @param eventType 事件类型
     */
    public static void deregisterPublisher(final Class<? extends Event> eventType) {
        final String topic = ClassUtils.getCanonicalName(eventType);
        EventPublisher publisher = INSTANCE.publisherMap.remove(topic);
        try {
            publisher.shutdown();
        } catch (Throwable ex) {
            log.error(">>> 😒 There was an exception when publisher shutdown : {}", ex);
        }
    }
}
